/*  
  -- The Lime II Project -- 

  A tuplespaces-based middleware for coordinating agents and mobile hosts.
  Copyright (C) 2005.
  Gian Pietro Picco, Amy L. Murphy, Lorenzo Bellini.

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
*/

package lime2;
import lime2.ca.*;
import java.util.Random;

/**
 * Manages the communication layer. The CommunicationMgr receives RemoteOps and processes them "in parallel" (thread 
 * pool): it uses a PriorityScheduler to create the abstraction of four different queues: "outgoing operations", 
 * "incoming operations", "outgoing events" and "incoming events" (in order of increasing priority of service). 
 * <br>
 * Operations model any operation on a remote LimeTupleSpace that requires a new operation to be started.
 * <br>
 * Events model the triggering of remote weak reactions.
 * <br>
 * Please note that RemoteOps that model an exchange of data that is part of a operation in progress are immediately
 * posted to the appropriate thread of the pool.
 *  
 * @author Gian Pietro Picco
 * @author Amy Murphy
 * @author Lorenzo Bellini 
 */

public class CommunicationMgr implements Runnable, LimeConstants
{  
 // queues: index constants	
 private static final int INCOMING_EVENTS_QUEUE     = 0; 
 private static final int OUTGOING_EVENTS_QUEUE     = 1;  
 private static final int INCOMING_OPERATIONS_QUEUE = 2;  
 private static final int OUTGOING_OPERATIONS_QUEUE = 3;  

 // queues: priority constants
 private static final int[] PRIORITY_FACTORS = new int[] {1, 1, 1, 1}; 	
 
 // singleton CommunicationMgr  
 private static CommunicationMgr cmgr = null;
 
 // thread pool
 private BufferedThreadPool pool;

 // priority scheduler 
 private PriorityScheduler pSch;
 
 // FragmentationProxy
 private ICommunicationAdapter ca;

 // implementations for Lime remote protocols
 private Protocols protocolsImpl;
 
 // this (long) counter is used to enforce uniqueness of operation identifiers
 static private long trids = -1;
 
 // the lock to synchronize access of trids
 static private Object tridLock = new Object();
 
 // this facility is used to generate the first sequence number of each operation 
 static private Random rnd = new Random();
 
 // private constructor
 private CommunicationMgr(String c) throws LimeCommunicationException 
 {
  // super	
  super();
  // creates thread pool
  pool = new BufferedThreadPool(POOLED_THREADS);
  // creates priority scheduler
  pSch = new PriorityScheduler(PRIORITY_FACTORS);
  // creates remote protocols
  protocolsImpl = new Protocols();
  // creates the communication adapter
  try 
  {
   ca = (ICommunicationAdapter) Class.forName(c).newInstance();	
   // + messages
   if (SHOW_MSG) ca = new MsgTestCA(ca);
   // + fragmentation
   if (FRAGMENT_OUTGOING_ENABLED || DEFRAGMENT_INCOMING_ENABLED) ca = new FragmentationProxy(ca, this);
  }
  catch (ClassNotFoundException cnfe) { throw new LimeCommunicationException("can't find the C-Adapter class", cnfe, Failures.CRITICAL); }
  catch (IllegalAccessException iae) { throw new LimeCommunicationException("can't load the C-Adapter class (is it public ?)", iae, Failures.CRITICAL); } 
  catch (InstantiationException ie) { throw new LimeCommunicationException("can't instantiate C-Adapter object", ie, Failures.CRITICAL); }
  // initializes the communication adapter
  ca.init();
 }
 
 /**
  * Builds the (singleton) communication manager. 
  * 
  * @param c name of a class implementing ICommunicationAdapter to use as communication adapter
  * @throws LimeCommunicationException
  */
 public static void build(String c) throws LimeCommunicationException
 {
  // instantiates the communication manager singleton	
  cmgr = new CommunicationMgr(c);
  // threading
  new Thread(cmgr).start();
  cmgr.new Poller().start();
 }
 
 /**
  * Returns the (singleton) CommunicationMgr.
  * 
  * @return CommunicationMgr
  */
 public static CommunicationMgr get()
 {
  return cmgr;	
 }
 
 /**
  * Returns the communication adapter.
  * 
  * @return communication adapter
  */
 public ICommunicationAdapter getCA()
 {
  return ca;	
 }

 /**
  * Returns the input buffer of a specific thread of the pool.
  * 
  * @param key identifier of the thread of the pool 
  * @return input buffer
  */
 IncomingBuffer getBuffer(byte key)
 {
  return pool.getBuffer(key);	
 }
 
 /**
  * Returns the input buffer of the current thread of the pool; <code>null</code> is returned if the invoker is not a
  * thread of the pool.
  * 
  * @return input buffer
  */
 IncomingBuffer getBuffer()
 {
  return pool.getBuffer();
 }
 
 /**
  * Returns an identifier of the current thread of the pool. This method is meaningless if it is not invoked by a 
  * thread of the pool.
  * 
  * @return identifier of the current thread
  */
 byte getThreadID()
 {
  return pool.getThreadID();	
 }
 
 /**
  * Adds a RemoteOp to the ougoing queue of operations.
  *  
  * @param o the RemoteOp (= RemTupleOp or RemReactionOp) to add to the outgoing queue of operations
  * @param system is the remote operation part of a reaction ?
  */
 public void postOp(RemoteOp o, boolean system)
 {
  if (system) serialRun(o);
  else pSch.schedule(OUTGOING_OPERATIONS_QUEUE, o);	
 }
 
 /**
  * Adds a RemoteOp to the outgoing queue of events. Events are always posted by the system: their addresser is a
  * remote peer - we don't need to force serial execution.
  *  
  * @param o the RemoteOp (= RemWeakOp / non-lazy RemReconcileOp) to add to the outgoing queue of events
  */
 public void postEvent(RemoteOp o)
 {
  pSch.schedule(OUTGOING_EVENTS_QUEUE, o);	
 }
 
 /**
  * Dispatches an incoming RemoteOp to the correct incoming queue.
  * 
  * @param o the incoming RemoteOp to dispatch
  */
 void dispatch(RemoteOp o)
 {
  // if this host is not engaged, (unicast tentative) requests from "remote" are simply discarded		
  if (!LimeServer.getServer().isEngaged()) return;	
 
  // the (unicast tentative) op comes from an host that is not yet engaged 
  if (!ca.queryHost(o.source)) ca.addHost(o.source);
  
  // records last tuple 
  GhostBag.saveLast(o);
  
  // discards duplicates	
  if (!DupCache.isDuplicate(o))
  {
   byte a = o.type;
   byte b = o.subtype;

   // data for an already started protocol
   if (b == REPLY || b == REJECT || b == ACK || b == FIN) pool.getBuffer(o.targetT).add(o); 
   else
   // PASSIVE_OPEN event (new protocol) + RECONCILE (.. we give high priority to reconciliation) 	
   if (a == REM_EVENTW || a == REM_RDW || a == REM_INW || a == RECONCILE) pSch.schedule(INCOMING_EVENTS_QUEUE, o); 
   else	
   // PASSIVE_OPEN operation (new protocol)	
   pSch.schedule(INCOMING_OPERATIONS_QUEUE, o); 
  } 
 }

 /**
	 * Returns the next unique identifier for operations. <br>
	 * The numeric identifier of the first operation started by this host is
	 * zero. Synchronization is necessary since multiple threads could try to
	 * access the global counter ("trids") when we are in the middle of a
	 * previous invocation of this method (.. this would ruin uniqueness of the
	 * operation identifiers).
	 * 
	 * @return unique operation identifier
	 */
 	long getTI() {
		synchronized (tridLock) {
			trids++;
			return trids;
		}
	}
 
 /**
	 * Return the first sequence number of a operation: it is a random number.
	 * During the operation it is incremented at each "step".
	 * 
	 * @return first sequence number of a operation
	 */
 int getSeqnum()
 {
  return rnd.nextInt();	
 }
 
 /**
  * Renews the timeout of all the buffers that are waiting for an op from the given tuplespace..  while reconciliation 
  * is performed the local tuplespace is temporarily "blocked" (we acquire the mutex of the corresponding LimeTSMgr). 
  * <br>
  * This is needed for reconciliation; consider the following <i>scenario</i>: host Y moves out of range just after 
  * sending an HELLO message to host X but before receiving an HELLO message from X; thus X considers itself still 
  * engaged with Y. Y obviously disagrees. Then X queries tuplespace T on Y for a tuple: Y will start the 
  * reconciliation protocol before replying to the query; X will start the "passive side" of reconciliation - but we 
  * don't want the timeout "on" the buffer of the original query to expire in the meanwhile.    
  * 
  * @param name of the tuplespace
  */
 void renewAll(String name)
 {
  IncomingBuffer[] ib = pool.getAllBuffers();	
  for (int i = 0; i < ib.length; i++)
  {
   RemoteOp aux = ib[i].getWaitingOp();
   if (aux != null && aux.name.equals(name)) ib[i].renew(); 
  }	
 }
 
 /**
  * Run method. We exploit the parallelism provided by the thread pool for requests by different agents.
  */
 public void run()
 {
  for (;;) { pool.assignTask(protocolsImpl.select((RemoteOp) pSch.serve())); }
 }

 /**
  * Operations requested within reactions must be executed sequentially by the Lime system; we must also protect 
  * against exceptions (.. see ProtocolTerminationException) in the communication protocols ("robustness").
  * 
  * @param ro the RemoteOp to serially execute
  */
 void serialRun(RemoteOp ro)
 {
//	System.out.println("*****Communication Manager: schedule reconciliation tuple from " + ro.source + " to: " + ro.target);
  try { protocolsImpl.select(ro).run(); } 
  catch (ProtocolTerminationException pte) { }
  catch (Exception e) { e.printStackTrace(); } 
 }
 
 /**
  * Retrieves the remote ops. Since the method "getNextOp" of ICommunicationAdapter is blocking, we require an extra
  * thread for polling. Please note that he communication adapter doesn't need to know anything about Lime and its
  * internal queues (maximum compatibility).
  */
 private class Poller extends Thread { public void run() { for(;;) dispatch(ca.getNextOp()); } }
   
}
